package com.atguigu.edu.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.beans.VideoViewerNumsBean;
import com.atguigu.edu.realtime.utils.DateFormatUtil;
import com.atguigu.edu.realtime.utils.MyClickhouseUtil;
import com.atguigu.edu.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.Collections;
import java.util.HashSet;

/**
 * @Classname DwsPlayViewerNums
 * @Date 2022/11/20 17:39
 * @Created by arun
 * 视频观看人数
 */
public class DwsPlayViewerNums {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

        String topic = "dwd_play_videoPlayTimes";
        String groupId = "dws_play_viewerNums_group";
        FlinkKafkaConsumer<String> kafkaConsumer = MyKafkaUtil.getKafkaConsumer(topic, groupId);
        DataStreamSource<String> kafkaStrDS = env.addSource(kafkaConsumer);
        // kafkaStrDS.print(">>>");
        // {"video_id":1014,"play_sec":"3946","position_sec":"mid_257","uid":"30","mid":"720",
        // "ts":"1668929324172","chapterId":21909}

        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaStrDS.map(JSON::parseObject);
        SingleOutputStreamOperator<VideoViewerNumsBean> numsBean = jsonObjDS.map(
                new MapFunction<JSONObject, VideoViewerNumsBean>() {
                    @Override
                    public VideoViewerNumsBean map(JSONObject jsonObj) throws Exception {
                        VideoViewerNumsBean numsBean = VideoViewerNumsBean.builder()
                                .videoId(jsonObj.getString("video_id"))
                                .userIdSet(new HashSet<>(Collections.singleton(jsonObj.getString("uid"))))
                                .ts(jsonObj.getLong("ts"))
                                .build();
                        return numsBean;
                    }
                }
        );

        SingleOutputStreamOperator<VideoViewerNumsBean> withWatermarkDS = numsBean.assignTimestampsAndWatermarks(
                WatermarkStrategy.<VideoViewerNumsBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner(new SerializableTimestampAssigner<VideoViewerNumsBean>() {
                            @Override
                            public long extractTimestamp(VideoViewerNumsBean element, long recordTimestamp) {
                                return element.getTs();
                            }
                        })
        );

        KeyedStream<VideoViewerNumsBean, String> keyedDS = withWatermarkDS.keyBy(VideoViewerNumsBean::getVideoId);

        WindowedStream<VideoViewerNumsBean, String, TimeWindow> windowDS = keyedDS.window(TumblingEventTimeWindows.of(Time.seconds(10)));

        SingleOutputStreamOperator<VideoViewerNumsBean> reduceDS = windowDS.reduce(
                new ReduceFunction<VideoViewerNumsBean>() {
                    @Override
                    public VideoViewerNumsBean reduce(VideoViewerNumsBean value1, VideoViewerNumsBean value2) throws Exception {
                        value1.getUserIdSet().addAll(value2.getUserIdSet());
                        return value1;
                    }
                },
                new WindowFunction<VideoViewerNumsBean, VideoViewerNumsBean, String, TimeWindow>() {
                    @Override
                    public void apply(String groupId, TimeWindow window, Iterable<VideoViewerNumsBean> input, Collector<VideoViewerNumsBean> out) throws Exception {
                        String stt = DateFormatUtil.toYmdHms(window.getStart());
                        String end = DateFormatUtil.toYmdHms(window.getEnd());
                        for (VideoViewerNumsBean numsBean : input) {
                            numsBean.setStt(stt);
                            numsBean.setEdt(end);
                            numsBean.setNums((long) numsBean.getUserIdSet().size());
                            out.collect(numsBean);
                        }
                    }
                }
        );
        reduceDS.print(">>>");

        reduceDS.addSink(
                MyClickhouseUtil.getSinkFunction("insert into dws_play_video_viewerNums_window  values(?,?,?,?,?)")
        );
        env.execute();
    }
}
